package com.grf.kafka.java.serializer;

import com.grf.kafka.java.KafkaConstant;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * @apiNote 用户生产者
 * @author grf
 * @since 2024/10/24
 */
@Slf4j
public class UserProduct {

    @SneakyThrows
    public static void main(String[] args) {
        //PART1:设置发送者相关属性
        Properties props = new Properties();
        // 此处配置的kafka的端口
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstant.BOOTSTRAP_SERVERS);
        // 配置key的序列化类
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"com.grf.kafka.java.serializer.UserSerializer");

        Producer<String, User> producer = new KafkaProducer<>(props);
        for (int i=0; i<10; i++) {
            // part2: 构建消息
            User user = new User();
            user.setId(i);
            user.setAge(i*10+8);
            user.setName("test"+i);
            user.setSex((i+1)%2);
            // log.info("user: {}", user);
            ProducerRecord<String, User> record = new ProducerRecord<>(KafkaConstant.TOPIC, Integer.toString(i), user);
            // part3: 发送消息
            // 单向发送，不关心服务应答
            // producer.send(record);
            // 同步发送，获取服务端应答消息前，会阻塞当前线程
            RecordMetadata metadata = producer.send(record).get();
            String topic = metadata.topic();
            int partition = metadata.partition();
            long offset = metadata.offset();
            String message = user.toString();
            log.info("topic is {}; partition is {}; offset is {}; message is {}", topic, partition, offset, message);
        }
    }
}
